package com.test.study.ToolsUtils.strom;

import java.util.Map;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * 产生数据
 * 
 * @author huyong
 *
 */
@SuppressWarnings("serial")
public class SpoutClass extends BaseRichSpout {
	@SuppressWarnings({ "unused", "rawtypes" })
	private Map conf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	private SpoutOutputCollector collector;

	/**
	 * 相当于初始化方法，当这个组件被初始化的时候执行一次，并且只会执行一次 参数说明：参数1:Storm的配置信息 参数2:storm上下文
	 * 参数3:storm的发射器
	 */
	@SuppressWarnings("rawtypes")
	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		this.conf = conf;
		this.context = context;
		this.collector = collector;
		System.out.println("conf===" + conf + "=====context===" + context + "====collector====" + collector);
	}

	int number = 0;

	/**
	 * 该方法是一个死循环，一直会被调用，不会停止 这里可以接受其他的数据源 比如消息队列 数据库
	 */
	@Override
	public void nextTuple() {
		try {
			Thread.sleep(500);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		// 这里传递一个messageID 相当于开启了消息确认机制 这个messageID相当于一个消息的主键ID messageid
		// 和tule之间的数据是一一对应的他们之间的关系需要程序员自己维护
		// 使用字段进行分组 分为基数和偶数
		this.collector.emit(new Values(number % 2, number++), number - 1);// 向下一个bolt发送数据 //数据变化 每次加1
	}

	/**
	 * 声明输出字段 方便bolt取数据
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("flag", "number"));// 注意：这个输出字段的个数与前面发射出去的个数是一一对应的，但是名称可以自定义
	}

//执行成功的回调函数
	@Override
	public void ack(Object msgId) {
		System.out.println("ack  msgId=" + msgId + "执行成功");
	}

//执行失败的回调函数
	@Override
	public void fail(Object msgId) {
		System.out.println("fail  msgId=" + msgId + "执行失败");
	}
	// 这里不管执行成功还是执行失败 回调函数中只能接受到这个消息的msgID 所以在使用的时候要将消息和msgid之间的对应关系应该保存下来

}
